#define __USE_GNU
#include <sched.h>
#include <sys/time.h>
#include <cstdlib>
#include "DolphinDB.h"
#include "Util.h"
#include <iostream>
#include <string>
#include <thread>
#include <vector>
#include <cinttypes>
#include <cstdio>
#include <map>
#include <dirent.h>
#include <unistd.h>
using namespace dolphindb;
using namespace std;

typedef struct {
    char sql[256];
    char dataDir[256];
    int filesNum;
    int clients;
    int rowsPerRequest;
    int write;
    int writeIn;
    char database[256];
    int read;
    bool dropDatabase;
    int port;
    char ipaddress[256];
} ProArgs;

typedef struct {
    int64_t totalRows;
} StatisInfo;

typedef struct {
    int threadId;
    int sID;
    int eID;
} ThreadObj;

static StatisInfo statis;
static ProArgs arguments;

void parseArg(int argc, char *argv[]);

void writeData();

void readData();

int main(int argc, char *argv[]) {
    statis.totalRows = 0;
    parseArg(argc, argv);

    if (arguments.write) {
        writeData();
    } else {
        readData();
    }
}

void parseArg(int argc, char *argv[]) {
    strcpy(arguments.sql, "./sqlCmd.txt");
    strcpy(arguments.dataDir, "./testdata");
    arguments.filesNum = 2;
    arguments.clients = 1;
    arguments.rowsPerRequest = 100;
    arguments.writeIn = 0;
    arguments.read = 1;
    arguments.dropDatabase = false;
    arguments.port = 8848;
    strcpy(arguments.ipaddress, "127.0.0.1");

    for (int i = 1; i < argc; ++i) {
        if (strcmp(argv[i], "-sql") == 0) {
            if (i < argc - 1) {
                strcpy(arguments.sql, argv[++i]);
            }
            else {
                fprintf(stderr, "'-sql' requires a parameter, default:%s\n", arguments.sql);
                exit(EXIT_FAILURE);
            }
        }
        else if (strcmp(argv[i], "-dataDir") == 0) {
            if (i < argc - 1) {
                strcpy(arguments.dataDir, argv[++i]);
            }
            else {
                fprintf(stderr, "'-dataDir' requires a parameter, default:%s\n", arguments.dataDir);
                exit(EXIT_FAILURE);
            }
        }
        else if (strcmp(argv[i], "-numOfFiles") == 0) {
            if (i < argc - 1) {
                arguments.filesNum = atoi(argv[++i]);
            }
            else {
                fprintf(stderr, "'-numOfFiles' requires a parameter, default:%d\n", arguments.filesNum);
                exit(EXIT_FAILURE);
            }
        }
        else if (strcmp(argv[i], "-clients") == 0) {
            if (i < argc - 1) {
                arguments.clients = atoi(argv[++i]);
            }
            else {
                fprintf(stderr, "'-clients' requires a parameter, default:%d\n", arguments.clients);
                exit(EXIT_FAILURE);
            }
        }
        else if (strcmp(argv[i], "-rowsPerRequest") == 0) {
            if (i < argc - 1) {
                arguments.rowsPerRequest = atoi(argv[++i]);
            }
            else {
                fprintf(stderr, "'-rowsPerRequest' requires a parameter, default:%d\n", arguments.rowsPerRequest);
                exit(EXIT_FAILURE);
            }
        }
        else if (strcmp(argv[i], "-w") == 0) {
            arguments.write = 1;
        }
        //1表示内存表，2表示分布式表，3表示读取整个文件，4表示整批写入内存表，5表示整批写入分布式表
        else if (strcmp(argv[i], "-writeIn") == 0) {
            if (i < argc - 1) {
                arguments.writeIn = atoi(argv[++i]);
            }
            else {
                fprintf(stderr, "'-writeIn' requires a parameter\n");
                exit(EXIT_FAILURE);
            }
        }
        else if (strcmp(argv[i], "-DatabasePos") == 0) {
            if (i < argc - 1) {
                strcpy(arguments.database,argv[++i]);
            }
            else {
                fprintf(stderr, "'-DatabasePos' requires a parameter\n");
                exit(EXIT_FAILURE);
            }
        }
        //1表示提前读入内存，2表示加入I/O时间，3表示进行一次性读入文件测试
        else if (strcmp(argv[i], "-read") == 0) {
            if (i < argc - 1) {
                arguments.read = atoi(argv[++i]);
            }
            else {
                fprintf(stderr, "'-read' requires a parameter\n");
                exit(EXIT_FAILURE);
            }
        }
        else if(strcmp(argv[i], "-dropDatabase") == 0){
            arguments.dropDatabase = true;
        }
        else if(strcmp(argv[i], "-port") == 0){
            if (i < argc - 1) {
                arguments.port = atoi(argv[++i]);
            }
            else {
                fprintf(stderr, "'-port' requires a integer\n");
                exit(EXIT_FAILURE);
            }
        }
        else if (strcmp(argv[i], "-ipaddress") == 0) {
            if (i < argc - 1) {
                strcpy(arguments.ipaddress,argv[++i]);
            }
            else {
                fprintf(stderr, "'-ipaddress' requires a parameter\n");
                exit(EXIT_FAILURE);
            }
        }
    }
}

int64_t getTimeStampMs() {
    struct timeval systemTime{};
    gettimeofday(&systemTime, nullptr);
    return (int64_t)systemTime.tv_sec * 1000L + (int64_t)systemTime.tv_usec / 1000;
}

void writeDataBatchRAM(void *param){
    auto *pThread = (ThreadObj *)param;
    //得到有关进程、线程和CPU的信息
    printf("Thread %d, writing sID %d, eID %d\n", pThread->threadId, pThread->sID, pThread->eID);

    cpu_set_t mask;
    CPU_ZERO(&mask);
    CPU_SET(pThread->threadId,&mask);
    if (sched_setaffinity(0, sizeof(mask), &mask) == -1){
        printf("CPU set failed\n");
        exit(EXIT_FAILURE);
    }

    __ssize_t result = 0;
    DBConnection conn;
    bool ret = conn.connect(arguments.ipaddress, arguments.port);
    if(!ret){
        cout<<"Failed to connect to the server"<<endl;
        return;
    }
    if(arguments.writeIn == 5) {
        conn.run("login(`admin,`123456);");
        conn.run("COLS_INFO=`devid`devname`devgroup`ts`temperature`humidity;");
        conn.run("COLS_TYPE=`INT`SYMBOL`INT`TIMESTAMP`INT`FLOAT;");
        conn.run("COLS_TYPE_FINAL=`INT`SYMBOL`INT`TIMESTAMP`INT`FLOAT;");
        conn.run("schematable=table(COLS_INFO as name, COLS_TYPE as type);");
        conn.run("TIME_RANGE=2019.07.16T04:00:00+1800*0..6;");
        conn.run(R"(ID_RANGE=("dev_"+lpad((0..500*1000+1)$STRING,7,"0")$SYMBOL);)");
        conn.run("time_schema=database('',RANGE,TIME_RANGE);");
        conn.run("id_schema=database('',RANGE,ID_RANGE);");
        conn.run("db=database(\""+ string(arguments.database) + "\",COMPO,[time_schema, id_schema]);");
        conn.run("tb=table(10000:0,COLS_INFO,COLS_TYPE_FINAL)");
    }

    //之后程序需要用到的参数
    char *sql = static_cast<char *>(calloc(1, 8 * 1024 * 1024));
    int init = 0;
    int counter = 0;
    int totalRecords = 0;

    vector<string> colNames = {"devid","devname","devgroup","ts","temperature","humidity"};
    vector<DATA_TYPE> colTypes = {DT_INT,DT_STRING,DT_INT,DT_TIMESTAMP,DT_INT,DT_FLOAT};
    int colNum = 6, rowNum = arguments.rowsPerRequest, indexCapacity=arguments.rowsPerRequest;
    vector<VectorSP> columnVecs;
    int length = 0;
    int64_t total_time = 0;

    //int64_t start = getTimeStampMs();
    for (int j = pThread->sID; j <= pThread->eID; j++) {
        //打开被指定的文件
        char fileName[300];
        sprintf(fileName, "%s/testdata%d.csv", arguments.dataDir, j);

        FILE *fp = fopen(fileName, "r");
        if (fp == nullptr) {
            printf("failed to open file %s\n", fileName);
            exit(1);
        }
        printf("open file %s success\n", fileName);

        char *line = nullptr;
        size_t len = 0;
        map<int,int> exist;
        ConstantSP table;

        init = 1;
        //一行一行地读文件
        while (!feof(fp)) {
            free(line);
            line = nullptr;
            len = 0;

            //读到一行
            result += getline(&line, &len, fp);
            if (line == nullptr) break;

            if (strlen(line) < 10) continue;
            int machineid;
            char machinename[16];
            int machinegroup;
            int64_t timestamp;
            int temperature;
            float humidity;
            sscanf(line, "%d%s%d%" PRId64 "%d%f", &machineid, machinename, &machinegroup, &timestamp, &temperature, &humidity);

            if(init){
                init = 0;
                table = Util::createTable(colNames, colTypes, rowNum, indexCapacity);
                columnVecs.clear();
                for(int i = 0; i < colNum; ++i)
                    columnVecs.emplace_back(table->getColumn(i));
                length = 0;
            }

            columnVecs[0]->setInt(length, machineid);
            columnVecs[1]->setString(length, string(machinename));
            columnVecs[2]->setInt(length, machinegroup);
            //columnVecs[3]->set(length, Util::createTimestamp(timestamp));
            columnVecs[3]->setLong(length,timestamp);
            columnVecs[4]->setInt(length, temperature);
            columnVecs[5]->setFloat(length++, humidity);

            counter++;

            int64_t st = getTimeStampMs();
            if (counter >= arguments.rowsPerRequest) {
                vector<ConstantSP> args;
                args.push_back(table);
                if(arguments.writeIn == 4)
                    conn.run("tableInsert{devices}",args);
                else if(arguments.writeIn == 5){
                    char temp[100];
                    sprintf(temp,"tableInsert{loadTable(\"%s\", `devices)}",string(arguments.database).c_str());
                    //cout << table->getString() << endl;
                    conn.run(temp,args);
                }

                totalRecords += counter;
                counter = 0;
                init = 1;
            }
            total_time+=getTimeStampMs()-st;
        }
        fclose(fp);
    }

    //int64_t et = getTimeStampMs();
    //cout << total_time/1000.0 << " " << (et-start)/1000.0 << " " << total_time*1.0/(et-start) << endl;
    __sync_fetch_and_add(&statis.totalRows, totalRecords);
    free(sql);
}

void writeDataRAM(void *param) {
    //得到有关进程、线程和CPU的信息
    auto *pThread = (ThreadObj *)param;
    printf("Thread %d, writing sID %d, eID %d\n", pThread->threadId, pThread->sID, pThread->eID);
    //cout << syscall(__NR_gettid) << endl;

    cpu_set_t mask;
    CPU_ZERO(&mask);
    CPU_SET(pThread->threadId,&mask);

    if (sched_setaffinity(0, sizeof(mask), &mask) == -1){
        printf("CPU set failed\n");
        exit(EXIT_FAILURE);
    }

    __ssize_t result = 0;
    DBConnection conn;
    bool ret = conn.connect(arguments.ipaddress, arguments.port);
    if(!ret){
        cout<<"Failed to connect to the server"<<endl;
        return;
    }
    if(arguments.writeIn == 3){
        FILE *fp = fopen("./q6.txt", "r");
        if (fp == nullptr) {
            printf("failed to open file %s\n", "./q6.txt");
            exit(1);
        }
        printf("open file %s success\n", "./q6.txt");

        char *line = nullptr;
        size_t len;
        while (!feof(fp)) {
            free(line);
            line = nullptr;
            len = 0;

            //读到一行
            result += getline(&line, &len, fp);
            conn.run(line);
        }
        fclose(fp);
        __sync_fetch_and_add(&statis.totalRows, 40000000);

        return;
    }

    //之后程序需要用到的参数
    char *sql = static_cast<char *>(calloc(1, 8 * 1024 * 1024));
    int sqlLen = 0;
    int counter = 0;
    int totalRecords = 0;

    for (int j = pThread->sID; j <= pThread->eID; j++) {
        //打开被指定的文件
        char fileName[300];
        sprintf(fileName, "%s/testdata%d.csv", arguments.dataDir, j);

        FILE *fp = fopen(fileName, "r");
        if (fp == nullptr) {
            printf("failed to open file %s\n", fileName);
            exit(1);
        }
        printf("open file %s success\n", fileName);

        char *line = nullptr;
        size_t len = 0;
        map<int,int> exist;

        //一行一行地读文件
        while (!feof(fp)) {
            free(line);
            line = nullptr;
            len = 0;

            //读到一行
            result = getline(&line, &len, fp);
            if (line == nullptr) break;

            if (strlen(line) < 10) continue;
            int machineid;
            char machinename[16];
            int machinegroup;
            int64_t timestamp;
            int temperature;
            float humidity;
            sscanf(line, "%d%s%d%" PRId64 "%d%f", &machineid, machinename, &machinegroup, &timestamp, &temperature, &humidity);

            //切换写到将数据写进数据库的目的地
            if (exist.find(machineid)==exist.end()) {
                exist[machineid] = 1;
                sqlLen += sprintf(sql + sqlLen, "dev_%d = table(10000:0, `devid`devname`devgroup`ts`temperature`humidity, [INT,STRING,INT,TIMESTAMP,INT,FLOAT]);",machineid);
                sqlLen += sprintf(sql + sqlLen, "share dev_%d as dev%d;",machineid, machineid);
            }

            sqlLen += sprintf(sql + sqlLen, "insert into dev%d values(%d,\"%s\",%d,timestamp(%" PRId64 "),%d,%f);",machineid,machineid,machinename,machinegroup,timestamp,temperature,humidity);
            counter++;

            if (counter >= arguments.rowsPerRequest) {
                conn.run(sql);

                totalRecords += counter;
                counter = 0;
                sqlLen = 0;
            }
        }
        fclose(fp);
    }

    if (counter > 0) {
        conn.run(sql);

        totalRecords += counter;
    }

    __sync_fetch_and_add(&statis.totalRows, totalRecords);
    free(sql);
}

void writeDataDFS(ThreadObj *pThread){
    //得到有关进程、线程和CPU的信息
    printf("Thread %d, writing sID %d, eID %d\n", pThread->threadId, pThread->sID, pThread->eID);

    cpu_set_t mask;
    CPU_ZERO(&mask);
    CPU_SET(pThread->threadId,&mask);
    if (sched_setaffinity(0, sizeof(mask), &mask) == -1){
        printf("CPU set failed\n");
        exit(EXIT_FAILURE);
    }

    __ssize_t result = 0;
    DBConnection conn;
    bool ret = conn.connect(arguments.ipaddress, arguments.port);
    if(!ret){
        cout<<"Failed to connect to the server"<<endl;
        return;
    }
    conn.run("db = database(\"" + string(arguments.database) + R"(", RANGE,("dev_"+lpad((0..100*100+1)$STRING,5,"0")$SYMBOL)))");

    //之后程序需要用到的参数
    char *sql = static_cast<char *>(calloc(1, 8 * 1024 * 1024));
    int sqlLen = 0;
    int lastMachineid = 0;
    int counter = 0;
    int totalRecords = 0;

    vector<TableSP> tables;
    vector<string> colNames = {"devid","devname","devgroup","ts","temperature","humidity"};
    vector<DATA_TYPE> colTypes = {DT_INT,DT_STRING,DT_INT,DT_TIMESTAMP,DT_INT,DT_FLOAT};
    int colNum = 6, rowNum = arguments.rowsPerRequest, indexCapacity=10000;
    vector<VectorSP> columnVecs;
    int length = 0;

    for (int j = pThread->sID; j <= pThread->eID; j++) {
        //打开被指定的文件
        char fileName[300];
        sprintf(fileName, "%s/testdata%d.csv", arguments.dataDir, j);

        FILE *fp = fopen(fileName, "r");
        if (fp == nullptr) {
            printf("failed to open file %s\n", fileName);
            exit(1);
        }
        printf("open file %s success\n", fileName);

        char *line = nullptr;
        size_t len = 0;
        map<int,int> exist;

        //一行一行地读文件
        while (!feof(fp)) {
            free(line);
            line = nullptr;
            len = 0;

            //读到一行
            result += getline(&line, &len, fp);
            if (line == nullptr) break;

            if (strlen(line) < 10) continue;
            int machineid;
            char machinename[16];
            int machinegroup;
            int64_t timestamp;
            int temperature;
            float humidity;
            sscanf(line, "%d%s%d%" PRId64 "%d%f", &machineid, machinename, &machinegroup, &timestamp, &temperature, &humidity);

            //切换写到将数据写进数据库的目的地
            if (exist.find(machineid) == exist.end()) {
                exist[machineid] = 1;
                sqlLen += sprintf(sql + sqlLen,
                                  "dev_%05d=db.createPartitionedTable(table(10000:0, `devid`devname`devgroup`ts`temperature`humidity, [INT,STRING,INT,TIMESTAMP,INT,FLOAT]), \"dev_%05d\", `devname);",
                                  machineid, machineid);
            }
            if(lastMachineid!=machineid){
                lastMachineid = machineid;
                tables.emplace_back(Util::createTable(colNames, colTypes, rowNum, indexCapacity));
                columnVecs.clear();
                for(int i = 0; i < colNum; ++i)
                    columnVecs.emplace_back(tables[tables.size()-1]->getColumn(i));
                length = 0;
            }

            columnVecs[0]->set(length, Util::createInt(machineid));
            columnVecs[1]->setString(length, string(machinename));
            columnVecs[2]->set(length, Util::createInt(machinegroup));
            columnVecs[3]->set(length, Util::createTimestamp(timestamp));
            columnVecs[4]->set(length, Util::createInt(temperature));
            columnVecs[5]->set(length++, Util::createFloat(humidity));

            counter++;

            if (counter >= arguments.rowsPerRequest) {
                if(sqlLen!=0)
                    conn.run(sql);
                for(auto & table : tables){
                    char temp[100];
                    vector<ConstantSP> args;
                    args.emplace_back(table);
                    sprintf(temp,"tableInsert{loadTable(\"%s\", `dev_%05d)}",string(arguments.database).c_str(),table->values()->get(0)->get(0)->getInt());
                    auto str_temp = string(temp);
                    //cout << args[0]->getString() << endl;
                    //cout << temp << endl;
                    conn.run(temp,args);
                }

                totalRecords += counter;
                counter = 0;
                sqlLen = 0;
                tables.clear();
                lastMachineid = -1;
            }
        }
        fclose(fp);
    }

    if (counter > 0) {
        if(sqlLen!=0)
            conn.run(sql);
        for(auto & table : tables){
            char temp[100];
            vector<ConstantSP> args;
            args.emplace_back(table);
            sprintf(temp,"tableInsert{loadTable(\"%s\", `dev_%05d)}",string(arguments.database).c_str(),table->values()->get(0)->get(0)->getInt());
            conn.run(temp,args);
        }

        tables.clear();
        totalRecords += counter;
    }

    __sync_fetch_and_add(&statis.totalRows, totalRecords);
    free(sql);
}

void writeData() {
    printf("write data\n");
    printf("---- clients: %d\n", arguments.clients);
    printf("---- dataDir: %s\n", arguments.dataDir);
    printf("---- numOfFiles: %d\n", arguments.filesNum);
    printf("---- rowsPerRequest: %d\n", arguments.rowsPerRequest);
    printf("---- writeIn: %d\n", arguments.writeIn);


    if(arguments.writeIn == 2) {
        DBConnection conn;
        bool ret = conn.connect(arguments.ipaddress, arguments.port);
        if (!ret) {
            cout << "Failed to connect to the server" << endl;
            return;
        }

        conn.run("db = database(\"" + string(arguments.database) + R"(", RANGE,("dev_"+lpad((0..100*100+1)$STRING,5,"0")$SYMBOL)))");
    }
    else if(arguments.writeIn == 4 || arguments.writeIn == 5){
        DBConnection conn;
        bool ret = conn.connect(arguments.ipaddress, arguments.port);
        if (!ret) {
            cout << "Failed to connect to the server" << endl;
            return;
        }
        if(arguments.writeIn == 4)
            conn.run("t=table(10:0, `devid`devname`devgroup`ts`temperature`humidity, [INT,STRING,INT,TIMESTAMP,INT,FLOAT]);\nshare t as devices;");
        else if(arguments.writeIn == 5){
            conn.run("login(`admin,`123456);");
            if (arguments.dropDatabase)
                conn.run("dropDatabase(\"" + string(arguments.database) + "\");");
            conn.run("COLS_INFO=`devid`devname`devgroup`ts`temperature`humidity;");
            conn.run("COLS_TYPE=`INT`SYMBOL`INT`TIMESTAMP`INT`FLOAT;");
            conn.run("COLS_TYPE_FINAL=`INT`SYMBOL`INT`TIMESTAMP`INT`FLOAT;");
            conn.run("schematable=table(COLS_INFO as name, COLS_TYPE as type);");
            conn.run("TIME_RANGE=2019.07.16T04:00:00+1800*0..6;");
            conn.run(R"(ID_RANGE=("dev_"+lpad((0..500*1000+1)$STRING,7,"0")$SYMBOL);)");
            conn.run("time_schema=database('',RANGE,TIME_RANGE);");
            conn.run("id_schema=database('',RANGE,ID_RANGE);");
            conn.run("db=database(\""+ string(arguments.database) + "\",COMPO,[time_schema, id_schema]);");
            conn.run("tb=table(10000:0,COLS_INFO,COLS_TYPE_FINAL)");
            conn.run("tb=db.createPartitionedTable(tb,`devices,`ts`devname);");
        }
    }

    int64_t st = getTimeStampMs();

    //arguments.filesNum-=36;
    int a = arguments.filesNum / arguments.clients;
    int b = arguments.filesNum % arguments.clients;
    int last = 0;


    vector<thread> threads;
    for (int i = 0; i < arguments.clients; ++i) {
        auto *pthread = new ThreadObj();
        pthread->threadId = i + 1;
        pthread->sID = last;
        if (i < b) {
            pthread->eID = last + a;
        } else {
            pthread->eID = last + a - 1;
        }
        last = pthread->eID + 1;
        if (i < b) {
            pthread->eID = pthread->sID + 2;
        } else {
            pthread->eID = pthread->sID + 1;
        }

        if(arguments.writeIn == 1 || arguments.writeIn == 3)
            threads.emplace_back(writeDataRAM,pthread);
        else if(arguments.writeIn == 2)
            threads.emplace_back(writeDataDFS,pthread);
        else if(arguments.writeIn == 4 || arguments.writeIn == 5)
            threads.emplace_back(writeDataBatchRAM,pthread);
        else{
            cout << "The writeIn should be in 1~5" << endl;
            exit(1);
        }
    }

    for (int i = 0; i < arguments.clients; i++) {
        threads[i].join();
    }

    int64_t elapsed = getTimeStampMs() - st;
    float seconds = (float)elapsed / 1000;
    float rs = (float)statis.totalRows / seconds;

    printf("---- Spent %f seconds to insert %" PRId64 " records, speed: %f Rows/Second\n", seconds, statis.totalRows, rs);
}

void readDataImp(ThreadObj *pThread)
{
    printf("Thread %d\n", pThread->threadId);

    cpu_set_t mask;
    CPU_ZERO(&mask);
    CPU_SET(pThread->threadId,&mask);

    if (sched_setaffinity(0, sizeof(mask), &mask) == -1){
        printf("CPU set failed\n");
        exit(EXIT_FAILURE);
    }

    FILE *fp = fopen(arguments.sql, "r");
    if (fp == nullptr) {
        printf("failed to open file %s\n", arguments.sql);
        exit(1);
    }
    printf("open file %s success\n", arguments.sql);

    __ssize_t result = 0;
    DBConnection conn;
    bool ret = conn.connect(arguments.ipaddress, arguments.port);
    if(!ret){
        cout<<"Failed to connect to the server"<<endl;
        return;
    }

    conn.run("login(`admin,`123456);");
    char temp_cmd[100*120];
    //sprintf(temp_cmd, "devices=loadTable(\"%s\",`devices,memoryMode=true);",arguments.database);
    sprintf(temp_cmd, "devices=loadTable(\"%s\",`devices);",arguments.database);

    if(arguments.read == 1) {
        int64_t st_pre = getTimeStampMs();
        conn.run(temp_cmd);
        int64_t elapsed_pre = getTimeStampMs() - st_pre;
        float seconds = (float) elapsed_pre / 1000;
        printf("---- Spent %f seconds to preload the data\n", seconds);
    }

    char *line = nullptr;
    size_t len = 0;
    while (!feof(fp)) {
        free(line);
        line = nullptr;
        len = 0;

        result += getline(&line, &len, fp);
        if (line == nullptr) break;

        if (strlen(line) < 10) continue;

        int64_t st = getTimeStampMs();
        if(arguments.read == 2)
            conn.run(temp_cmd);
        ConstantSP table = conn.run(line);

        int64_t elapsed = getTimeStampMs() - st;
        float seconds = (float)elapsed / 1000;
        printf("---- Spent %f seconds to retrieve %d records, Thread:%d query: %s\n", seconds, table->size(), pThread->threadId, line);
    }

    fclose(fp);
}

void readData() {
    printf("read data\n");
    printf("---- sql: %s\n", arguments.sql);
    printf("---- clients: %d\n", arguments.clients);

    vector<thread> threads;
    for (int i = 0; i < arguments.clients; ++i) {
        auto *pthread = new ThreadObj();
        pthread->threadId = i + 1;
        threads.emplace_back(readDataImp,pthread);
    }

    for (int i = 0; i < arguments.clients; i++) {
        threads[i].join();
    }
}
